package app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import app.bean.ContentInfo;
import app.bean.ContentRankResult;
import app.bean.ContentSalesAccumulator;
import app.function.ContentRankProcessFunction;
import utils.MyClickHouseUtil;
import utils.MyKafkaUtil;
import utils.SystemConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 内容排行榜处理器 - 符合实时数仓设计规范
 * 实现标准化的DWS层数据处理流程
 */
public class DwsMarketContentRank {
    
    private static final Logger logger = LoggerFactory.getLogger(DwsMarketContentRank.class);
    private static final String DATA_SOURCE = "kafka_dwd_market_content";
    private static final Integer VERSION = 1;
    
    public static void main(String[] args) throws Exception {
        // 初始化系统配置，避免Hadoop路径问题
        SystemConfigUtil.initializeSystemConfig();
        
        logger.info("启动DwsMarketContentRank...");
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "dwd_market_content";
        String groupId = "dws_market_content_rank_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换为ContentInfo对象并添加水位线
        SingleOutputStreamOperator<ContentInfo> contentDS = kafkaDS.map(new MapFunction<String, ContentInfo>() {
            @Override
            public ContentInfo map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                ContentInfo content = new ContentInfo();
                content.setContentId(jsonObject.getString("content_id"));
                content.setContentTitle(jsonObject.getString("title"));
                content.setContentType(jsonObject.getString("content_type"));
                content.setShopId(jsonObject.getString("shop_id"));
                content.setShopName(jsonObject.getString("shop_name"));
                content.setViewCount(jsonObject.getLong("view_count"));
                content.setClickCount(jsonObject.getLong("click_count"));
                content.setOrderCount(jsonObject.getLong("order_count"));
                content.setSalesAmount(jsonObject.getDouble("sales_amount"));
                content.setConversionRate(jsonObject.getDouble("conversion_rate"));
                return content;
            }
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<ContentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<ContentInfo>() {
                    @Override
                    public long extractTimestamp(ContentInfo element, long recordTimestamp) {
                        return System.currentTimeMillis();
                    }
                })
        );
        
        // 按内容ID分组
        KeyedStream<ContentInfo, String> keyedStream = contentDS.keyBy(new KeySelector<ContentInfo, String>() {
            @Override
            public String getKey(ContentInfo value) throws Exception {
                return value.getContentId();
            }
        });
        
        // 1. 计算直播交易榜
        SingleOutputStreamOperator<ContentRankResult> liveTransactionRankDS = contentDS
            .filter(content -> "live".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new LiveTransactionAggregateFunction(), new LiveTransactionWindowFunction())
            .keyBy(content -> "live_transaction")
            .process(new ContentRankProcessFunction("live_transaction", 100));
        
        // 2. 计算直播流量榜
        SingleOutputStreamOperator<ContentRankResult> liveTrafficRankDS = contentDS
            .filter(content -> "live".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new LiveTrafficAggregateFunction(), new LiveTrafficWindowFunction())
            .keyBy(content -> "live_traffic")
            .process(new ContentRankProcessFunction("live_traffic", 100));
        
        // 3. 计算短视频带货榜
        SingleOutputStreamOperator<ContentRankResult> videoTransactionRankDS = contentDS
            .filter(content -> "video".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new VideoTransactionAggregateFunction(), new VideoTransactionWindowFunction())
            .keyBy(content -> "video_transaction")
            .process(new ContentRankProcessFunction("video_transaction", 100));
        
        // 4. 计算短视频引流榜
        SingleOutputStreamOperator<ContentRankResult> videoClickRankDS = contentDS
            .filter(content -> "video".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new VideoClickAggregateFunction(), new VideoClickWindowFunction())
            .keyBy(content -> "video_click")
            .process(new ContentRankProcessFunction("video_click", 100));
        
        // 5. 计算热门短视频榜
        SingleOutputStreamOperator<ContentRankResult> videoViewRankDS = contentDS
            .filter(content -> "video".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new VideoViewAggregateFunction(), new VideoViewWindowFunction())
            .keyBy(content -> "video_view")
            .process(new ContentRankProcessFunction("video_view", 100));
        
        // 6. 计算带货图文榜
        SingleOutputStreamOperator<ContentRankResult> articleTransactionRankDS = contentDS
            .filter(content -> "article".equals(content.getContentType()))
            .keyBy(ContentInfo::getContentId)
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new ArticleTransactionAggregateFunction(), new ArticleTransactionWindowFunction())
            .keyBy(content -> "article_transaction")
            .process(new ContentRankProcessFunction("article_transaction", 100));
        
        // 将结果写入ClickHouse - 使用符合实时数仓规范的SQL
        liveTransactionRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_live_transaction_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        liveTrafficRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_live_traffic_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        videoTransactionRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_video_transaction_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        videoClickRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_video_click_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        videoViewRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_video_view_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        articleTransactionRankDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_content_article_transaction_rank VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        // 执行任务
        env.execute("Content Ranking Job");
    }
    
    // 直播交易聚合函数
    public static class LiveTransactionAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.totalClickCount += value.getClickCount();
            accumulator.totalOrderCount += value.getOrderCount();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            accumulator.calculateAvgConversionRate();
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.totalClickCount += b.totalClickCount;
            a.totalOrderCount += b.totalOrderCount;
            a.totalSalesAmount += b.totalSalesAmount;
            a.count += b.count;
            return a;
        }
    }
    
    // 直播交易窗口函数
    public static class LiveTransactionWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setViewCount(accumulator.totalViewCount);
            result.setClickCount(accumulator.totalClickCount);
            result.setOrderCount(accumulator.totalOrderCount);
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            
            // 计算转化率
            result.calculateConversionRate();
            result.calculateClickRate();
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 直播流量聚合函数
    public static class LiveTrafficAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.totalClickCount += value.getClickCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.totalClickCount += b.totalClickCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 直播流量窗口函数
    public static class LiveTrafficWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段 - 主要关注流量
            result.setViewCount(accumulator.totalViewCount);
            result.setClickCount(accumulator.totalClickCount);
            
            // 计算点击率
            result.calculateClickRate();
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 短视频带货聚合函数
    public static class VideoTransactionAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.totalClickCount += value.getClickCount();
            accumulator.totalOrderCount += value.getOrderCount();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            accumulator.calculateAvgConversionRate();
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.totalClickCount += b.totalClickCount;
            a.totalOrderCount += b.totalOrderCount;
            a.totalSalesAmount += b.totalSalesAmount;
            a.count += b.count;
            return a;
        }
    }
    
    // 短视频带货窗口函数
    public static class VideoTransactionWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setViewCount(accumulator.totalViewCount);
            result.setClickCount(accumulator.totalClickCount);
            result.setOrderCount(accumulator.totalOrderCount);
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            
            // 计算转化率
            result.calculateConversionRate();
            result.calculateClickRate();
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 短视频引流聚合函数
    public static class VideoClickAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.totalClickCount += value.getClickCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.totalClickCount += b.totalClickCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 短视频引流窗口函数
    public static class VideoClickWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段 - 主要关注点击量
            result.setViewCount(accumulator.totalViewCount);
            result.setClickCount(accumulator.totalClickCount);
            
            // 计算点击率
            result.calculateClickRate();
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 短视频观看量聚合函数
    public static class VideoViewAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.count += b.count;
            return a;
        }
    }
    
    // 短视频观看量窗口函数
    public static class VideoViewWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段 - 主要关注观看量
            result.setViewCount(accumulator.totalViewCount);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 图文带货聚合函数
    public static class ArticleTransactionAggregateFunction implements AggregateFunction<ContentInfo, ContentSalesAccumulator, ContentSalesAccumulator> {
        @Override
        public ContentSalesAccumulator createAccumulator() {
            return new ContentSalesAccumulator();
        }
        
        @Override
        public ContentSalesAccumulator add(ContentInfo value, ContentSalesAccumulator accumulator) {
            accumulator.contentId = value.getContentId();
            accumulator.contentTitle = value.getContentTitle();
            accumulator.contentType = value.getContentType();
            accumulator.shopId = value.getShopId();
            accumulator.shopName = value.getShopName();
            accumulator.totalViewCount += value.getViewCount();
            accumulator.totalClickCount += value.getClickCount();
            accumulator.totalOrderCount += value.getOrderCount();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator getResult(ContentSalesAccumulator accumulator) {
            accumulator.calculateAvgConversionRate();
            return accumulator;
        }
        
        @Override
        public ContentSalesAccumulator merge(ContentSalesAccumulator a, ContentSalesAccumulator b) {
            a.totalViewCount += b.totalViewCount;
            a.totalClickCount += b.totalClickCount;
            a.totalOrderCount += b.totalOrderCount;
            a.totalSalesAmount += b.totalSalesAmount;
            a.count += b.count;
            return a;
        }
    }
    
    // 图文带货窗口函数
    public static class ArticleTransactionWindowFunction implements WindowFunction<ContentSalesAccumulator, ContentRankResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<ContentSalesAccumulator> input, Collector<ContentRankResult> out) throws Exception {
            ContentSalesAccumulator accumulator = input.iterator().next();
            
            ContentRankResult result = new ContentRankResult();
            // 设置业务字段
            result.setContentId(accumulator.contentId);
            result.setContentTitle(accumulator.contentTitle);
            result.setContentType(accumulator.contentType);
            result.setShopId(accumulator.shopId);
            result.setShopName(accumulator.shopName);
            
            // 设置度量字段
            result.setViewCount(accumulator.totalViewCount);
            result.setClickCount(accumulator.totalClickCount);
            result.setOrderCount(accumulator.totalOrderCount);
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            
            // 计算转化率
            result.calculateConversionRate();
            result.calculateClickRate();
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
}